/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.pipe.CloseState;
import com.nhncorp.neptune.changelog.pipe.Pipe;

public class WorkerPool {
  static long lastReportTime = 0l;
  static final Log LOG = LogFactory.getLog(WorkerPool.class);
  List<Worker> workerList = new ArrayList<Worker>();
  int port;

  WorkerPool(int size, int port) throws IOException {
    this.port = port;

    for (int i = 0; i < size; i++) {
      Worker worker = new Worker(this);
      worker.setName("Worker#" + (i + 1) + " on " + port);
      workerList.add(worker);
      worker.start();
    }
  }

  public Worker getWorker() throws IOException {
    int minimum = Integer.MAX_VALUE;
    Worker idleWorker = null;
    
//    long curTime = System.currentTimeMillis();
//    if (lastReportTime == 0 || curTime - ChangeLogServerIF.REPORT_TIME > lastReportTime) {
//      report();
//      lastReportTime = System.currentTimeMillis();
//    }

    for (Worker worker : workerList) {
      int pipeCount = worker.getPipeCount();
      if (pipeCount == 0) {
        return worker;
      } else if (minimum > pipeCount) {
        minimum = pipeCount;
        idleWorker = worker;
      }
    }

    return idleWorker;
  }

  void report() {
    for (Worker worker : workerList) {
      LOG.info("worker[" + worker.getName() + "] pipe count : " + worker.pipeSet.size());
    }
  }

  public void shutdown() {
    for (Worker worker : workerList) {
      worker.shutdown();
    }
  }

}

class Worker extends Thread {
  static final Log LOG = LogFactory.getLog(Worker.class);

  AtomicBoolean doSelectingBlock = new AtomicBoolean(true);
  Selector selector;
  LinkedList<Pipe> newPipeList = new LinkedList<Pipe>();
  ReentrantLock pipeSetLock = new ReentrantLock();
  Set<Pipe> pipeSet = new TreeSet<Pipe>();
  int currentIndex = -1;

  boolean shutdown = false;

  private WorkerPool workerPool;

  Worker(WorkerPool workerPool) throws IOException {
    this.workerPool = workerPool;
    selector = Selector.open();
  }

  public void shutdown() {
    shutdown = true;
    LOG.debug(this.getName() + " is interrupted. Current num of pipes are " + pipeSet.size());
    this.interrupt();
    selector.wakeup();
  }

  public int getPipeCount() {
    pipeSetLock.lock();
    try {
      return pipeSet.size();
    } finally {
      pipeSetLock.unlock();
    }
  }

  long startTime = 0;
  long processTime = 0;
  long cleanupTime = 0;
  
  public void run() {
    try {
      int numSelected = 0;
      while (true) {
        checkNewPipeEntry();
        
        if (shutdown) {
          break;
        }
        
        long endTime = System.currentTimeMillis();
        // 전체 4초 이상, 이벤트 처리하는데 100ms이상 걸린 경우에 report
        if (startTime > 0 &&  endTime - startTime >= 4000 && processTime - startTime > 100) {
          LOG.warn("The time taken for " + getName() + " to process events exceeds 4 sec. total : " 
              + (endTime - startTime) 
              + "ms, process : " + (processTime - startTime) 
              + "ms, cleanup : " + (cleanupTime - processTime) 
              + "ms, check : " + (endTime - cleanupTime) + "ms.");
          LOG.info("numSelectedEvents : " + numSelected);
          workerPool.report();
        }

        try {
          if (doSelectingBlock.get()) {
            numSelected = selector.select(500);
          } else {
            numSelected = selector.selectNow();
          }
        } catch (IOException e) {
          LOG.warn("Exception in selecting events", e);
        }

        startTime = System.currentTimeMillis();
        
        if (numSelected > 0) {
          dispatchSelectedKeys();
        }

        processTime = System.currentTimeMillis();

        cleanupExpiredPipes(processTime - startTime);
        
        cleanupTime = System.currentTimeMillis();
      }

      cleanupAll();
      LOG.debug(this.getName() + " is done");
    } catch (Throwable e) {
      LOG.warn("Worker [" + getName() + "] exits unexpectedly due to ", e);
      cleanupAll();
      System.gc();
      LOG.info("FREE : " + Runtime.getRuntime().freeMemory() + " Bytes, TOTAL : " + Runtime.getRuntime().totalMemory() + "Bytes");

      try {
        Worker worker = new Worker(workerPool);
        worker.setName(getName());
        workerPool.workerList.add(worker);
        worker.start();
      } catch (IOException ex) {
        LOG.warn("fail to fork new worker thread", ex);
      }
    }
  }

  private void dispatchSelectedKeys() {
    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();

    while (iter.hasNext()) {
      SelectionKey key = iter.next();
      iter.remove();

      if (key.isValid()) {
        if (key.isReadable()) {
          handleData(key);
        } else if (key.isConnectable()) {
          handleConnection(key);
        } else if (key.isWritable()) {
          handleWrite(key);
        }
      }
    }
  }

  private void handleWrite(SelectionKey key) {
    Pipe pipe = (Pipe) key.attachment();

    try {
      pipe.write(key.channel());
    } catch (Exception e) {
      LOG.error("error in handling write due to : ", e);
      cleanupErrorPipe(pipe);
    }
  }

  private void handleConnection(SelectionKey key) {
    Pipe pipe = (Pipe) key.attachment();

    try {
      pipe.connect(key.channel());
    } catch (Exception e) {
      LOG.error("error in handling connection due to : ", e);
      cleanupErrorPipe(pipe);
    }
  }

  private void handleData(SelectionKey key) {
    Pipe pipe = (Pipe) key.attachment();

    try {
      pipe.read(key.channel());
    } catch (Exception e) {
      if (e instanceof PipeNormallyClosed) {
        pipe.close();
        LOG.debug("Pipe#" + pipe.getPipeKey() + " is gracefully closed.");
        pipeSetLock.lock();
        try {
          pipeSet.remove(pipe);
        } finally {
          pipeSetLock.unlock();
        }
      } else {
        LOG.error("error in reading data due to ", e);
        cleanupErrorPipe(pipe);
      }
    }
  }

  // // cleanup class and methods ////

  private void cleanupExpiredPipes(long processTime) {
    pipeSetLock.lock();
    try {
      Iterator<Pipe> iter = pipeSet.iterator();
      while (iter.hasNext()) {
        Pipe pipe = iter.next();

        if (pipe.isExpired(processTime)) {
          LOG.warn("Pipe#" + pipe + " is closed since it is expired.");
          iter.remove();
          pipe.close();
        } else if (pipe.isCloseTriggered() && pipe.isCloseState() == false) {
          try {
            pipe.changeState(CloseState.instance());
          } catch (IOException e) {
            LOG.warn("exception in changing to close state", e);
          }
        }
      }
    } finally {
      pipeSetLock.unlock();
    }
  }

  private void cleanupAll() {
    LOG.debug("CLEAN UP ALL THE PIPES");
    pipeSetLock.lock();
    try {
      for (Pipe pipe : pipeSet) {
        pipe.close();
      }

      pipeSet.clear();
    } finally {
      pipeSetLock.unlock();
    }

    try {
      selector.close();
    } catch (IOException e) {
      LOG.warn("closing selector is fail", e);
    }
  }

  private void cleanupErrorPipe(Pipe pipe) {
    LOG.info("clean up error pipe : " + pipe);
    pipe.close();

    pipeSetLock.lock();
    try {
      pipeSet.remove(pipe);
    } finally {
      pipeSetLock.unlock();
    }
  }

  private void checkNewPipeEntry() {
    pipeSetLock.lock();
    try {
      if (pipeSet.isEmpty()) {
        synchronized (newPipeList) {
          while (newPipeList.isEmpty()) {
            try {
              LOG.debug("Waiting for new pipe");
              pipeSetLock.unlock();
              newPipeList.wait();
              pipeSetLock.lock();
            } catch (InterruptedException e) {
              pipeSetLock.lock();
              return;
            }
          }
          registerNewPipe();
        }
      } else {
        synchronized (newPipeList) {
          if (newPipeList.isEmpty() == false) {
            registerNewPipe();
          }
        }
        doSelectingBlock.set(true);
      }
    } finally {
      pipeSetLock.unlock();
    }
  }
  
  long lastCheckedTime;
  int lastSizeOfPipeSet;
  // checkNewPipeEntry에서 pipeSetLock을 획득한 채로 호출된다.
  private void registerNewPipe() {
    while (newPipeList.size() > 0) {
      Pipe pipe = newPipeList.removeFirst();
      // LOG.debug("Thr " + Thread.currentThread().getName()
      // + " starts processing pipe#" + pipe.getPipeKey());
      pipe.register(selector);

      try {
        pipe.init();
      } catch (IOException e) {
        LOG.warn("initalizing pipe is fail", e);
        return;
      }

      if (pipeSet.add(pipe) == false) {
        LOG.error("Error in adding pipe to pipeSet since the set has already same pipe!");
        pipe.close();
      }
      checkNumOfPipes();
    }
  }

  private void checkNumOfPipes() {
    long curTime = System.currentTimeMillis();
    
    if ((curTime - lastCheckedTime) > 60000) {
      lastCheckedTime = curTime;
      if (lastSizeOfPipeSet * 2 > pipeSet.size()) {
        LOG.info("The num of pipes are doubled since " + new Date(lastCheckedTime));
      }
      lastSizeOfPipeSet = pipeSet.size();
      LOG.info("num of pipes : " + lastSizeOfPipeSet);
    }
  }

  public void append(Pipe pipe) {
    synchronized (newPipeList) {
      newPipeList.addFirst(pipe);
      newPipeList.notify();
    }
    selector.wakeup();
    doSelectingBlock.set(false);
  }
}
